//package com.tjhk.demo.kafka.listener;
//
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.ConsumerRecords;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.stereotype.Component;
//
//@Component
//public class KafkaBatchConsumer {
//
//    private static final Logger log= LoggerFactory.getLogger(KafkaBatchConsumer.class);
//
//
//    @KafkaListener(topics = "tjg-cyc-location")
//    public void batchListener(ConsumerRecords<String, byte[]> records, Acknowledgment ack){
//        log.info(">>>>>>.TreadID:{},接收消息数量：{}条", Thread.currentThread().getId(), records.count());
//        for (ConsumerRecord<String, byte[]> record : records) {
//            try {
//                log.info("partition={}, offset={}, key ={},value ={}", record.partition(), record.offset(), record.key(), new String(record.value(),"GBK"));
//            } catch (Exception e) {
//                log.error("丢弃消息，数据格式转换错误！-->partition={}, offset={}, key ={},value ={}", record.partition(), record.offset(), record.key(), record.value());
//            }
//        }
//        // 手动提交
//        ack.acknowledge();
//    }
//
//}
//
